package com.fitness.fitness.kafka;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.fitness.fitness.modal.datavisual.entity.CourseTopNVisual;
import com.fitness.fitness.modal.resource.entity.Venue;
import com.fitness.fitness.service.IVenueService;
import com.fitness.fitness.ws.Message;
import com.fitness.fitness.ws.WebSocketServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

@Slf4j
@Component
public class RealVenueListener {

    @Resource
    private IVenueService venueService;

    private Map<Long,Venue> venues;
    private Map<Long,Map<String,Object>> venueData = new HashMap<>();
    @PostConstruct
    public void init(){
        venues = venueService.getVenueMap();
    }

    @KafkaListener(topics = "venueRealMsg", groupId = "realGroup")
    public void listen(String msg) {
        if(StrUtil.isBlank(msg)){
            log.warn("no message!!!");
            return;
        }

        log.info("input value: {}" , msg);

        try {
            JSONObject jsonObject = JSONUtil.parseObj(msg);
            Long venueId = jsonObject.get("venueId",Long.class);
            Integer num = jsonObject.get("num",Integer.class);
            Venue venue = venues.get(venueId);
            String venueName = venue.getName();
            String address = venue.getAddress();
            Map<String,Object> venueMsg = new HashMap<>();
            //添加名称
            venueMsg.put("venueName",venueName);
            //添加人数
            venueMsg.put("num",num);
            //添加地址
            venueMsg.put("address",StrUtil.isBlank(address) ? "-" : address);
            //添加更新时间
            venueMsg.put("timestamp_",DateUtil.now());
            venueData.put(venueId, venueMsg);

            Message<List<Object>> message = new Message<>();
            message.setType("venueRealVisual");
            message.setContent(Arrays.asList(venueData.values().stream().sorted(new Comparator<Map<String, Object>>() {
                @Override
                public int compare(Map<String, Object> o1, Map<String, Object> o2) {
                    return o2.get("timestamp_").toString().compareTo(o1.get("timestamp_").toString());
                }
            }).toArray()));
            WebSocketServer.sendAllMessage(JSONUtil.toJsonStr(message));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
